package cn.doit.etl

import cn.doit.common.DolphinAppComm
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}


/**
  * 将bz2日志文件转换成parquet文件
  */
object Bz2Parquet {
  def main(args: Array[String]): Unit = {
    //校验参数
    if(args.length!=2){
      println(
        """
          |Usage:cn.doit.File2Parquet
          |Param:
          |   bz2InputPath     日志输入路径
          |   parquetOutPath    parquet输出路径
        """.stripMargin)
      sys.exit(-1)  //-1是非正常退出
    }

    //接收参数(模式匹配)
    val Array(bz2InputPath,parquetOutPath)=args

    val sc: SparkContext = DolphinAppComm.creatSparkContext("将日志转换成parquet")


//读取离线的bz2日志文件
    val file = sc.textFile(bz2InputPath)
//过滤非法数据
    val filteRDD = file.map(_.split(",",-1)).filter(_.size>=85)
//创建SQL实例
    val sqlc = new SQLContext(sc)

//导入隐式转换(把钥匙给他)


    import cn.doit.bean.RichString._

    val rowRDD = filteRDD.map(arr => Row(
      arr(0),
      arr(1).toIntPlus,
      arr(2).toIntPlus,
      arr(3).toIntPlus,
      arr(4).toIntPlus,
      arr(5),
      arr(6),
      arr(7).toIntPlus,
      arr(8).toIntPlus,
      arr(9).toDoublePlus,
      arr(10).toDoublePlus,
      arr(11),
      arr(12),
      arr(13),
      arr(14),
      arr(15),
      arr(16),
      arr(17).toIntPlus,
      arr(18),
      arr(19),
      arr(20).toIntPlus,
      arr(21).toIntPlus,
      arr(22),
      arr(23),
      arr(24),
      arr(25),
      arr(26).toIntPlus,
      arr(27),
      arr(28).toIntPlus,
      arr(29),
      arr(30).toIntPlus,
      arr(31).toIntPlus,
      arr(32).toIntPlus,
      arr(33),
      arr(34).toIntPlus,
      arr(35).toIntPlus,
      arr(36).toIntPlus,
      arr(37),
      arr(38).toIntPlus,
      arr(39).toIntPlus,
      arr(40).toDoublePlus,
      arr(41).toDoublePlus,
      arr(42).toIntPlus,
      arr(43),
      arr(44).toDoublePlus,
      arr(45).toDoublePlus,
      arr(46),
      arr(47),
      arr(48),
      arr(49),
      arr(50),
      arr(51),
      arr(52),
      arr(53),
      arr(54),
      arr(55),
      arr(56),
      arr(57).toIntPlus,
      arr(58).toDoublePlus,
      arr(59).toIntPlus,
      arr(60).toIntPlus,
      arr(61),
      arr(62),
      arr(63),
      arr(64),
      arr(65),
      arr(66),
      arr(67),
      arr(68),
      arr(69),
      arr(70),
      arr(71),
      arr(72),
      arr(73).toIntPlus,
      arr(74).toDoublePlus,
      arr(75).toDoublePlus,
      arr(76).toDoublePlus,
      arr(77).toDoublePlus,
      arr(78).toDoublePlus,
      arr(79),
      arr(80),
      arr(81),
      arr(82),
      arr(83),
      arr(84).toIntPlus
    ))

//schema:
val schema = StructType(Seq(
  StructField("sessionid", StringType),
  StructField("advertisersid", IntegerType),
  StructField("adorderid", IntegerType),
  StructField("adcreativeid", IntegerType),
  StructField("adplatformproviderid", IntegerType),
  StructField("sdkversion", StringType),
  StructField("adplatformkey", StringType),
  StructField("putinmodeltype", IntegerType),
  StructField("requestmode", IntegerType),
  StructField("adprice", DoubleType),
  StructField("adppprice", DoubleType),
  StructField("requestdate", StringType),
  StructField("ip", StringType),
  StructField("appid", StringType),
  StructField("appname", StringType),
  StructField("uuid", StringType),
  StructField("device", StringType),
  StructField("client", IntegerType),
  StructField("osversion", StringType),
  StructField("density", StringType),
  StructField("pw", IntegerType),
  StructField("ph", IntegerType),
  StructField("long", StringType),
  StructField("lat", StringType),
  StructField("provincename", StringType),
  StructField("cityname", StringType),
  StructField("ispid", IntegerType),
  StructField("ispname", StringType),
  StructField("networkmannerid", IntegerType),
  StructField("networkmannername",StringType),
  StructField("iseffective", IntegerType),
  StructField("isbilling", IntegerType),
  StructField("adspacetype", IntegerType),
  StructField("adspacetypename", StringType),
  StructField("devicetype", IntegerType),
  StructField("processnode", IntegerType),
  StructField("apptype", IntegerType),
  StructField("district", StringType),
  StructField("paymode", IntegerType),
  StructField("isbid", IntegerType),
  StructField("bidprice", DoubleType),
  StructField("winprice", DoubleType),
  StructField("iswin", IntegerType),
  StructField("cur", StringType),
  StructField("rate", DoubleType),
  StructField("cnywinprice", DoubleType),
  StructField("imei", StringType),
  StructField("mac", StringType),
  StructField("idfa", StringType),
  StructField("openudid", StringType),
  StructField("androidid", StringType),
  StructField("rtbprovince", StringType),
  StructField("rtbcity", StringType),
  StructField("rtbdistrict", StringType),
  StructField("rtbstreet", StringType),
  StructField("storeurl", StringType),
  StructField("realip", StringType),
  StructField("isqualityapp", IntegerType),
  StructField("bidfloor", DoubleType),
  StructField("aw", IntegerType),
  StructField("ah", IntegerType),
  StructField("imeimd5", StringType),
  StructField("macmd5", StringType),
  StructField("idfamd5", StringType),
  StructField("openudidmd5", StringType),
  StructField("androididmd5", StringType),
  StructField("imeisha1", StringType),
  StructField("macsha1", StringType),
  StructField("idfasha1", StringType),
  StructField("openudidsha1", StringType),
  StructField("androididsha1", StringType),
  StructField("uuidunknow", StringType),
  StructField("userid", StringType),
  StructField("iptype", IntegerType),
  StructField("initbidprice", DoubleType),
  StructField("adpayment", DoubleType),
  StructField("agentrate", DoubleType),
  StructField("lrate", DoubleType),
  StructField("adxrate", DoubleType),
  StructField("title", StringType),
  StructField("keywords", StringType),
  StructField("tagid", StringType),
  StructField("callbackdate", StringType),
  StructField("channelid", StringType),
  StructField("mediatype", IntegerType)
))


val dataFrame = sqlc.createDataFrame(rowRDD,schema)

  dataFrame.write.parquet(parquetOutPath)

    sc.stop()

  }
}
